<?php
/**
 * Created by PhpStorm.
 * User: zgh
 * Date: 2021-09-03
 * Time: 16:42
 */

namespace app\common\command;


use app\common\model\MessageModel;
use app\common\util\RedisUtil;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;

/**
 * 任务队列 使用  消息订阅进行处理
 * Class TaskQueueCommand
 * @package app\common\command
 */
class TaskQueueCommand extends Command
{

    protected $subscribe_arr = ['order'];

    const STREAM_ORDER_KEY = 'stream_order';

    const STREAM_ORDER_GROUP = 'group_order'; //普通商品订单消费组


    public function execute(Input $input, Output $output)
    {
        ini_set("default_socket_timeout", -1); //保证redis 连接不超时
        if (memory_get_usage() > 100 * 1024 * 1024) { //保证内存不会产生泄露
            exit(0);
        }
        $redisUtil = new  RedisUtil();
        $redis = $redisUtil->getHandler();
        //查询是否已经创建了消费组  如果没有就创建nd("xino", "group");
        $group_arr = $redis->rawCommand("xinfo", "groups", self::STREAM_ORDER_KEY);
        if (empty($group_arr)) {
            //从0开始创建
            $redis->rawCommand("xgroup", "create", self::STREAM_ORDER_KEY, self::STREAM_ORDER_GROUP, "0");
        }
        $flag = true;
        while ($flag) {
            //下查询有没有 还没有ack的消息  有就优先处理这些还没有ack的消息  然后再进行读取新的消息
            $message_arr = [];
            if ($no_ack_dev_arr = $redis->rawCommand('xpending', self::STREAM_ORDER_KEY, self::STREAM_ORDER_GROUP, '-', '+', 10)) {
                foreach ($no_ack_dev_arr as $no_ack_row) {
                    $message_arr[] = $redis->rawCommand("xrange", self::STREAM_ORDER_KEY, $no_ack_row[0], "+", "count", 1)[0];
                }
            } else {
                $getResult = $redis->rawCommand("xreadgroup", "group", self::STREAM_ORDER_GROUP, "customA", "block", "2000", "count", 1, "streams", self::STREAM_ORDER_KEY, ">");
                if ($getResult) {
                    foreach ($getResult as $item) {
                        if ($item[0] == self::STREAM_ORDER_KEY) {
                            $message_arr = $item[1];
                        }
                    }
                }
            }
            $success = 0;
            if ($message_arr) {
                $data = [];
                //将一维数组 转化为 二维数组
                foreach ($message_arr as $row) {
                    $tmp = [];
                    $msg_row = $row[1];
                    $length = count($msg_row);
                    for ($ii = 0; $ii < $length; $ii += 2) {
                        $even = $ii + 1;
                        $tmp[$msg_row[$ii]] = $msg_row[$even];
                    }
                    $data[$row[0]] = $tmp;
                }
                foreach ($data as $msg_id => $msg_item) {
                    try {
                        Db::startTrans();
                        //处理业务完成 进一步在业务里做校验层 防止错误
                        $db_trans_res = true;
                        if ($db_trans_res) {
                            $model = new MessageModel();
                            $model->save([
                                "user_id" => $msg_item['user_id'],
                                "title" => "测试"
                            ]);
                            //确定消费成功
                            $ack_res = $redis->rawCommand("xack", self::STREAM_ORDER_KEY, self::STREAM_ORDER_GROUP, $msg_id);
                            if (empty($ack_res)) { //确认消息失败
                                Db::rollback();
                            }
                            Db::commit();
                            $success++;
                        } else {
                            Db::rollback();
                        }
                    } catch (\Exception $exception) {
//                        $exception->getMessage();
                        Db::rollback();
                    }
                }
            }
            echo '共' . count($message_arr) . "条消息,成功消费" . $success;
            sleep(1);//休眠1秒  订单量大的时候就进行调整该值
        }
    }


    protected function configure()
    {
        $this->setName('task_queue');
    }


    //使用消息订阅脚本进行
//    protected function execute(Input $input, Output $output)
//    {
//        ini_set("default_socket_timeout", -1); //保证redis 连接不超时
//        if (memory_get_usage() > 100 * 1024 * 1024) { //保证内存不会产生泄露
//            exit(0);
//        }
//        $redis = new \Redis();
//        $redis->connect('127.0.0.1', 16379);
//        $redis->subscribe($this->subscribe_arr, function ($redis, $chan, $msg) {
//            switch ($chan) {
//                case 'order':
//                    $task = unserialize($msg);
//                    if (!($user_id = $task['user_id']) || !($good_id = $task['good_id'])) {
//                        exit("参数错误");
//                    }
//                    try {
//                        Db::startTrans();
//                        //开始处理订单
//                        $good_info = Db::name("goods")->where(["status" => 1, "id" => $good_id])->find();
//                        if (!$good_info) {
//                            echo '没找到相关商品';
//                            Db::rollback();
//                            return false;
//                        }
//                        if ($good_info['stock'] <= 0) {
//                            echo '商品库存不足';
//                            Db::rollback();
//                            return false;
//                        }
//                        //查询有没有下单
//                        $order = Db::name("orders")->where(["goods_id" => $good_id, "user_id" => $user_id])->find();
//                        if (!empty($order)) {
//                            echo '请勿重复下单';
//                            Db::rollback();
//                            return false;
//                        }
//                        //创建订单 直接模拟支付成功
//                        $order_no = date("YmdHis") . mt_rand(10000, 99999);
//                        $time = time();
//                        $create_order_res = DB::name("orders")->insert([
//                            'order_no' => $order_no,
//                            'goods_id' => $good_id,
//                            'user_id' => $user_id,
//                            'pay_price' => $good_info['sell_price'],
//                            'is_pay' => 1,
//                            'pay_time' => $time, //支付时间
//                            'status' => 1,
//                            'create_time' => $time,
//                            'update_time' => $time
//                        ]);
//                        if (!$create_order_res) {
//                            echo '下单失败';
//                            Db::rollback();
//                            return false;
//                        }
//                        //减库存
//                        $stock = max(0, $good_info['stock'] - 1);
//                        $desc_stock_res = Db::name("goods")->where(["id" => $good_id])->update(['stock' => $stock, "update_time" => $time]);
//                        if (!$desc_stock_res) {
//                            echo '更新库存失败';
//                            Db::rollback();
//                            return false;
//                        }
//                        Db::commit();
//                        // 2代表的是抢购成功
//                        /**
//                         * @var $redis \Redis
//                         */
//                        //这里要重新开一个出来操作 否则操作返回都是false
//                        //这里重新开一个对象不能监听消息的时候 该进程做其他东西
//                        $new_redis = new \Redis();
//                        $new_redis->connect('127.0.0.1', 16379);
//                        $new_redis->hSet(RedisUtil::getPrefix() . "goods:" . $good_id, "userid_" . $user_id, 2);
//                        echo '下单成功';
//                        return true;
//                    } catch (\Exception $exception) {
//                        Db::rollback();
//                        exit();
//                    }
//                    break;
//                default:
//                    break;
//            }
//        });
//        exit;
//    }


}